package org.atlantis.datasink.mx;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class EventMonitor implements EventMonitorMBean {

	private AtomicLong totalMessageSize = new AtomicLong(0);
	private AtomicLong discardMessageSize = new AtomicLong(0);
	private String name;
	private int messageLimit;
	private int currentMessageSize;
	private int workerLimit;
	private int currentWorkerSize;
	private FragmentMonitor fragMonitor;

	public EventMonitor(String name, int messageLimit, int workerLimit) {
		this.name = name;
		this.messageLimit = messageLimit;
		this.workerLimit = workerLimit;
		this.fragMonitor = new FragmentMonitor(TimeUnit.SECONDS.toNanos(10));
	}

	@Override
	public String getMBeanName() {
		return "datasink:type=datasink.event." + name;
	}

	@Override
	public String getName() {
		return name;
	}

	@Override
	public int getMessageLimit() {
		return messageLimit;
	}

	@Override
	public int getCurrentMessageSize() {
		return currentMessageSize;
	}

	public void setCurrentMessageSize(int currentMessageSize) {
		this.currentMessageSize = currentMessageSize;
	}

	@Override
	public int getWorkerLimit() {
		return workerLimit;
	}

	@Override
	public int getCurrentWorkerSize() {
		return currentWorkerSize;
	}

	public void setCurrentWorkerSize(int currentWorkerSize) {
		this.currentWorkerSize = currentWorkerSize;
	}

	@Override
	public long getTotalMessageSize() {
		return totalMessageSize.get();
	}

	public void incTotalMessageSize() {
		fragMonitor.record();
		totalMessageSize.incrementAndGet();
	}

	@Override
	public long getDiscardMessageSize() {
		return discardMessageSize.get();
	}

	public void incDiscardMessageSize() {
		discardMessageSize.incrementAndGet();
	}

	@Override
	public double getSpeedBySec() {
		return fragMonitor.getSpeedBySec();
	}

	@Override
	public double getSpeedByMs() {
		return fragMonitor.getSpeedByMs();
	}
}
